package com.atguigu.test_02;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;



/*
* 算子链：将满足条件的算子链接到一起，形成一个大的task，该操作称之为算子链合并。
*
* 满足条件：
*       上下游的并行度一致，且数据的分发规则为forward
*
* 数据的分发规则（分区规则）
*       RescalePartitioner: 相对负载均衡。
*       rebalancePartitioner:  绝对负载均衡，上游的数据会轮询发送到下游的算子
*       KeyGroupStreamPartitioner: hash，按照key的hash对下游并行度取余
*       GlobalPartitioner: 强制并行度为1，数据全部发往下游的第一个并行度中
*       ForwardPartitioner: 直连，要求上游的并行度必须一致
*       ShufflePartitioner: 洗牌，随机发送下游的并行度
*       BroadcastPartitioner: 广播，将数据广播到下游的每个并行度
*       CustomPartitioner:
*
*禁用算子链：
*   1.针对与某个算子
*       startNewChain(): 从当前算子开始一个新链（不和上游算子合并，和下游算子合并）
*       disableChaining(): 当前算子禁用算子链（不与上下游合并）
*   2.针对于全局
*       env.disableOperatorChaining();
*
*算子链的好处：
*   它减少线程间切换、缓冲的开销，并且减少延迟的同时增加整体吞吐量。
*
* */

public class OperatorChain {
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",5678);
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //设置并行度
        env.setParallelism(1);

        //读取数据

        DataStreamSource<String> Ds = env.socketTextStream("hadoop102",8888);

        Ds.rebalance()
            .map(x -> x).name("map1")
                .map(x -> x).name("map2")
                        .map(x -> x).name("map3")
                                .print();

        // 启动执行
        env.execute();
    }
}
